/*********************************************************
          File Name:worker.c
          Author: Abby Cin
          Mail: abbytsing@gmail.com
          Created Time: Mon 27 Jul 2015 03:17:08 PM CST
**********************************************************/

#include "worker.h"
#include "event.h"
#include "sock.h"
#include "signals.h"
#include <assert.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <stdlib.h>
#include <sys/socket.h>

//#include <sys/epoll.h>

#ifdef  DBUG

#include <sched.h>
#define dbug(x)         do {\
        fprintf(stderr, "\n--%d--\t%s\n", x, strerror(errno));      \
}while(0);

#else
#define dbug(x)         {}
#endif

static struct task *job = NULL;
static int thread_num = 0;
static int breakall = 0;

static void sigterm_handler(int sig)
{
        int i;
        void *res;
        for(i = 0; i < thread_num; i++)
        {
                pthread_mutex_lock(&job[i].lock);
                job[i].exit = 1;
                job[i].work = 1;
                pthread_cond_signal(&job[i].cond);
                pthread_mutex_unlock(&job[i].lock);
        }

        for(i = 0; i < thread_num; i++)
        {
                if((errno = pthread_join(job[i].id, &res)) != 0)
                        perror("pthread_join()");
                else
                {
                        close(job[i].sockfd);  // ignore error
                        pthread_mutex_destroy(&job[i].lock);
                        pthread_rwlock_destroy(&job[i].rw);
                        pthread_cond_destroy(&job[i].cond);
                        
                }
        }

        breakall = 1;
}

static void worker_thread(void *arg)
{
        struct task *thread = (struct task *)arg;

        for(;;)
        {
                pthread_mutex_lock(&thread->lock);
                while(thread->work == 0)
                        pthread_cond_wait(&thread->cond, &thread->lock);
                thread->work = 0;

                if(thread->exit == 1)
                        break;

                // replace to your own codes
                //sleep(1);
                if(write(thread->sockfd, "back", 5) == -1)
                        dbug(__LINE__)
                if(close(thread->sockfd) == -1)
                        dbug(__LINE__)
                pthread_rwlock_wrlock(&thread->rw);
                thread->done = 1;
                pthread_rwlock_unlock(&thread->rw);
                pthread_mutex_unlock(&thread->lock);
        }
        pthread_mutex_unlock(&thread->lock);
        //pthread_exit(NULL);                   // cause memory leak
}

static int create_thread(int n, struct task *task,
                pthread_mutexattr_t *attr, pthread_rwlockattr_t *attr1)
{
        int i, res = 0;

        for(i = 0; i < n; i++)
        {
                task[i].sockfd = 0;
                task[i].exit = 0;
                task[i].done = 1;               // idle
                task[i].work = 0;
                pthread_mutex_init(&task[i].lock, attr);
                pthread_cond_init(&task[i].cond, NULL);
                pthread_rwlock_init(&task[i].rw, attr1);

                res = pthread_create(&task[i].id, NULL, (void *)worker_thread,
                                (void *)&task[i]);
                if(res != 0)
                        return -1;
        }

        return 0;
}

static void ev_cb(struct task *job, int sockfd)
{
        pthread_mutex_lock(&job->lock);
        job->done = 0;
        job->sockfd = sockfd;
        job->work = 1;
        pthread_cond_signal(&job->cond);
        pthread_mutex_unlock(&job->lock);
}

static void wroker_process(int n, int listenfd,int cpu, int sched)
{
#ifdef __linux__
        if(sched)
        {
                cpu_set_t set;
                CPU_ZERO(&set);
                CPU_SET(cpu, &set);
                if(sched_setaffinity(getpid(), sizeof(set), &set) == -1)
                        ;       // write log
                CPU_CLR(cpu, &set);
        }
#ifdef DBUG
        printf("%d is running on CPU %d\n", getpid(), sched_getcpu());
#endif
#endif
        install_signal_handler(SIGTERM, sigterm_handler);

        struct task local_job[n];
        memset(local_job, 0, n * sizeof(struct task));
        job = local_job;
        thread_num = n;
        Event *events = NULL;
        int efd, res;
        pthread_mutexattr_t attr;
        pthread_rwlockattr_t attr1;

        pthread_mutexattr_init(&attr);
        pthread_rwlockattr_init(&attr1);
        pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_DEFAULT);

        if(create_thread(n, job, &attr, &attr1) == -1)
                dbug(__LINE__)

        efd = ev_init(listenfd);
        if(efd == -1)
                dbug(__LINE__)

        events = (Event *)calloc(MAX_EVENTS, sizeof(Event));
        assert(events);

        res = ev_run(&breakall, listenfd, &efd, events, n, job, ev_cb);
        if(res == -1)
                ;// write log
        pthread_mutexattr_destroy(&attr);
        pthread_rwlockattr_destroy(&attr1);
        close(listenfd);
        ev_done(&efd, events);
}

static pid_t get_pid(int thread, int listenfd, int cpu, int sched)
{
        pid_t pid;

        switch((pid = fork()))
        {
                case -1:
                        return -1;
                case 0:
                        wroker_process(thread, listenfd, cpu, sched);
                        printf("\e[35m%d is exiting\e[0m\n", getpid());
                        exit(0);
                default:
                        break;
        }

        return pid;
}

int create_worker(int process, int thread, int listenfd, pid_t *result)
{
        int i, sched;
        int cpus = sysconf(_SC_NPROCESSORS_ONLN);
        printf("CPU online %d\n", cpus);

        sched = process == 1 ? 0 : 1;

        for(i = 0; i < process; i++)
        {
                result[i] = get_pid(thread, listenfd, (i+1) % cpus, sched);
                if(result[i] == -1)
                        return -1;
        }

        return 0;
}
